package com.atguigu.gmall.app.dws.examination;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.bean.DwsExaminationByCurriculumBena;
import com.atguigu.gmall.bean.DwsExaminationByCurriculumBenaThree;
import com.atguigu.gmall.bean.DwsExaminationByCurriculumBenaTwo;
import com.atguigu.gmall.bean.ExaminationBeanTwo;
import com.atguigu.gmall.utils.*;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.sql.SQLException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

public class DwsExaminationByCurriculum {
    public static void main(String[] args) throws SQLException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env);
        String topicName = "topic_db";
        String groupID = "DwsExaminationByCurriculum";
        streamTableEnvironment.executeSql(" create table topic_db( " +
                " `database` string, " +
                " `table` string, " +
                " `type` string, " +
                " `data` map<string,string>, " +
                " `ts` bigint ) "
                + SQLUtil.getKafkaSource(topicName,groupID));

        Table table = streamTableEnvironment.sqlQuery(
                        " select " +
                        " data['id'] id, " +
                        " data['paper_id'] paper_id, " +
                        " data['user_id'] user_id, " +
                        " cast(data['score'] as bigint) score, " +
                        " cast(data['duration_sec'] as bigint) duration_sec," +
                        " ts " +
                        " from topic_db " +
                        " where `table` = 'test_exam' " +
                        " and `type` = 'insert'" );

        DataStream<DwsExaminationByCurriculumBena> dwsExaminationByCurriculumDataStream = streamTableEnvironment.toAppendStream(table, DwsExaminationByCurriculumBena.class);
        SingleOutputStreamOperator<DwsExaminationByCurriculumBenaTwo> map = dwsExaminationByCurriculumDataStream.map(new MapFunction<DwsExaminationByCurriculumBena, DwsExaminationByCurriculumBenaTwo>() {
            @Override
            public DwsExaminationByCurriculumBenaTwo map(DwsExaminationByCurriculumBena value) throws Exception {

                String id = value.getPaper_id();
                DruidDataSource dataSource = DruidPhoenixDSUtil.getDataSource();
                List<JSONObject> dimInfo = DimUtil.getDimInfo(dataSource.getConnection(), "DIM_TEST_PAPER", new Tuple2<>("id", id));
                String courseId = dimInfo.get(0).get("courseId").toString();
                return new DwsExaminationByCurriculumBenaTwo(null, null, value.getId(), courseId, value.getUser_id(), value.getScore(), value.getDuration_sec(),0,value.getTs()*1000);
            }
        });

        KeyedStream<DwsExaminationByCurriculumBenaTwo, String> dwsExaminationByCurriculumBenaTwoStringKeyedStream1 = map.keyBy(bean -> bean.getCourse_id()+bean.getUser_id());
        SingleOutputStreamOperator<DwsExaminationByCurriculumBenaTwo> usercount = dwsExaminationByCurriculumBenaTwoStringKeyedStream1.process(new ProcessFunction<DwsExaminationByCurriculumBenaTwo, DwsExaminationByCurriculumBenaTwo>() {

            ValueState<String> usercount;

            @Override
            public void open(Configuration parameters) throws Exception {
                usercount = getRuntimeContext().getState(new ValueStateDescriptor<String>("usercount", String.class));
            }

            @Override
            public void processElement(DwsExaminationByCurriculumBenaTwo value, Context ctx, Collector<DwsExaminationByCurriculumBenaTwo> out) throws Exception {
                String i =value.getCourse_id() + value.getUser_id();
                if (usercount.value() == null) {
                    value.setUserCount(value.getUserCount() + 1);
                    usercount.update(i);
                    out.collect(value);
                }else {
                    out.collect(value);
                }

            }
        });

        KeyedStream<DwsExaminationByCurriculumBenaTwo, String> dwsExaminationByCurriculumBenaTwoStringKeyedStream = usercount.assignTimestampsAndWatermarks
                (WatermarkStrategy.<DwsExaminationByCurriculumBenaTwo>forBoundedOutOfOrderness(Duration.ofSeconds(2l)).withTimestampAssigner((bean, ts) -> bean.getTs()))
                .keyBy(bean -> bean.getCourse_id());
        SingleOutputStreamOperator<DwsExaminationByCurriculumBenaThree> reduce = dwsExaminationByCurriculumBenaTwoStringKeyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(3L)))
                .reduce(new ReduceFunction<DwsExaminationByCurriculumBenaTwo>() {
                    @Override
                    public DwsExaminationByCurriculumBenaTwo reduce(DwsExaminationByCurriculumBenaTwo value1, DwsExaminationByCurriculumBenaTwo value2) throws Exception {
                        value1.setUserCount(value1.getUserCount()+ value2.getUserCount());
                        value1.setScore(value1.getScore() + value2.getScore());
                        value1.setDuration_sec(value1.getDuration_sec() + value2.getDuration_sec());
                        return value1;
                    }
                }, new WindowFunction<DwsExaminationByCurriculumBenaTwo, DwsExaminationByCurriculumBenaThree, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<DwsExaminationByCurriculumBenaTwo> input, Collector<DwsExaminationByCurriculumBenaThree> out) throws Exception {
                        DwsExaminationByCurriculumBenaTwo next = input.iterator().next();
                        DwsExaminationByCurriculumBenaThree dwsExaminationByCurriculumBenaThree = new DwsExaminationByCurriculumBenaThree();
                        dwsExaminationByCurriculumBenaThree.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        dwsExaminationByCurriculumBenaThree.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        dwsExaminationByCurriculumBenaThree.setUserIdCount(next.getUserCount());
                        dwsExaminationByCurriculumBenaThree.setCourse_id(next.getCourse_id());
                        dwsExaminationByCurriculumBenaThree.setScore(next.getScore()/next.getUserCount());
                        dwsExaminationByCurriculumBenaThree.setDuration_sec(next.getDuration_sec()/next.getUserCount());
                        dwsExaminationByCurriculumBenaThree.setTs(next.getTs());
                        out.collect(dwsExaminationByCurriculumBenaThree);
                    }
                });
        reduce.print();

        String sql = "insert into dws_examination_curriculum values(?,?,?,?,?,?,?)";
        reduce.addSink(ClickHouseUtil.<DwsExaminationByCurriculumBenaThree>getJdbcSink(sql));
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
